package com.shujia.dwi

import com.shujia.utils.SparkTool
import com.shujia.utils.Config
import java.text.SimpleDateFormat

import com.shujia.grld.Grid

/**
  * 停留表计算
  *
  * 对融合表的数据进行合并，将同一个人在同一个网格中的数据合并成一条数据
  * 合并之后后面的任务，每一次读取的数据量减少，
  *
  */
object StayPointJob extends SparkTool {
  /**
    * 子类实现此方法，实现具体的代码逻辑
    *
    */
  override def run(args: Array[String]): Unit = {
    if (args.length == 0) {
      log.error("请指定时间参数：")
      //抛出异常
      throw new RuntimeException("请指定时间参数：day_id")
    }
    val day_id = args(0)
    log.info(s"当前处理的时间分区为：$day_id")


    /**
      * 读取融合表数据
      *
      */

    val mergeLocationPath = Config.get("mergeLocation.path") + "day_id=" + day_id

    val mergeLocationData = sc.textFile(mergeLocationPath)


    /**
      * 合并同一个人在同一个网格中的数据
      * 1、切分字段
      * 2、根据手机号和网格编号分组
      * 3、对组内地而数据进行合并
      * 4、整理数据将结果保存到hdfs
      *
      */

    //1、切分字段
    val kvData = mergeLocationData.map(line => {
      val split = line.split("\t")
      val mdn = split(0)
      val start_time = split(1)
      val county_id = split(2)
      val grid_id = split(6)

      val key = mdn + "-" + grid_id + "-" + county_id
      val value = start_time

      (key, value)
    })

    //2、根据手机号和网格编号分组
    val groupByData = kvData.groupByKey(20)


    //3、对组内地而数据进行合并

    val stayPointData = groupByData.map(kv => {
      val mdnAndGrid = kv._1

      val mdn = mdnAndGrid.split("-")(0)
      val grid_id = mdnAndGrid.split("-")(1)
      val county_id = mdnAndGrid.split("-")(2)


      val points = kv._2

      //开始时间
      var startTIme = Long.MaxValue

      //结束时间
      var endTIme = Long.MinValue


      /**
        * 循环所有时间，去时间的最大值和最小值
        *
        */
      for (point <- points) {
        val time = point.split(",")(0).toLong

        if (time < startTIme) {
          startTIme = time
        }

        if (time > endTIme) {
          endTIme = time
        }
      }

      val gridFirstTime = startTIme.toString
      val gridLastTime = endTIme.toString


      //计算网格停留时间，通过时间戳计算
      val dataFormat = new SimpleDateFormat("yyyyMMddHHmmss")
      val durations = dataFormat.parse(gridLastTime).getTime - dataFormat.parse(gridFirstTime).getTime
      //分钟级别的停留时间
      val duration = durations / 60000.0


      //去除经纬度  117400032945040
      val point = Grid.getCenter(grid_id.toLong)
      val longi = point.x
      val lati = point.y


      //保留小数点后四位
      val longiStr = "%.4f".format(longi)
      val latiStr = "%.4f".format(lati)



      // 整理返回的数据格式
      s"$mdn\t$longiStr\t$latiStr\t$grid_id\t$county_id\t$duration\t$gridFirstTime\t$gridLastTime"
    })

    val stayPointPath = Config.get("staypoint.path") + "day_id=" + day_id

    //保存数据
    save(stayPointData, stayPointPath)


  }

  /**
    * 指定spark配置
    * conf.set(key,value)
    */
  override def init(): Unit = {
    //    conf.setMaster("local")
  }
}
